package com.jcj.jcjrpc.core.consumer;

import com.jcj.jcjrpc.core.api.*;
import com.jcj.jcjrpc.core.consumer.http.OkHttpInvoker;
import com.jcj.jcjrpc.core.governance.SlidingTimeWindow;
import com.jcj.jcjrpc.core.meta.InstanceMeta;
import com.jcj.jcjrpc.core.util.MethodUtils;
import com.jcj.jcjrpc.core.util.TypeUtils;
import lombok.extern.slf4j.Slf4j;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.net.SocketTimeoutException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

@Slf4j
public class JcjInvocationHandler implements InvocationHandler {

    Class<?> service;
    RpcContext context;
    HttpInvoker httpInvoker;
    final List<InstanceMeta> providers;

    final List<InstanceMeta> isolatedProviders = new ArrayList<>();

    final List<InstanceMeta> halfOpenProviders = new ArrayList<>();

    final Map<String, SlidingTimeWindow> windows = new HashMap<>();
    ScheduledExecutorService executor;


    public JcjInvocationHandler(Class<?> clazz, RpcContext context, List<InstanceMeta> providers) {
        this.service = clazz;
        this.context = context;
        this.providers = providers;
        int timeout = context.getConsumerProperties().getTimeout();

        this.httpInvoker = new OkHttpInvoker(timeout);

        this.executor = Executors.newScheduledThreadPool(1);

        int halfOpenInitialDelay = context.getConsumerProperties().getHalfOpenInitialDelay();
        int halfOpenDelay = context.getConsumerProperties().getHalfOpenDelay();
        this.executor.scheduleWithFixedDelay(this::halfOpen, halfOpenInitialDelay,
                halfOpenDelay, TimeUnit.MILLISECONDS);
    }

    /**
     * 探活 每一分钟加入一次。
     */
    private void halfOpen() {
        log.debug(" ====> half open isolatedProviders: " + isolatedProviders);

        halfOpenProviders.clear();
        halfOpenProviders.addAll(isolatedProviders);
    }


    @Override
    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {

        if (MethodUtils.checkLocalMethod(method.getName())) {
            return null;
        }
        RpcRequest request = new RpcRequest();
        request.setService(service.getCanonicalName());
        request.setMethodSign(MethodUtils.methodSign(method));
        request.setArgs(args);

        int faultLimit = context.getConsumerProperties().getFaultLimit();
        int retries = context.getConsumerProperties().getRetries();
        while (retries-- > 0) {
            log.debug(" ===> reties: " + retries);
            try {
                //过滤器 prefilter
                for (Filter filter : this.context.getFilters()) {
                    Object preResult = filter.prefilter(request);
                    if (preResult != null) {
                        log.debug(filter.getClass().getName() + " ==> prefilter: " + preResult);
                        return preResult;
                    }
                }

                InstanceMeta instance;
                synchronized (halfOpenProviders) {
                    if (halfOpenProviders.isEmpty()) {
                        List<InstanceMeta> instances = context.getRouter().route(providers);
                        instance = context.getLoadBalancer().choose(instances);

                        log.debug(" loadBalancer.choose(instances) ==> {}", instance);

                    } else {
                        instance = halfOpenProviders.remove(0);
                        log.debug(" check alive instance ==> {}", instance);
                    }

                }

                RpcResponse<?> rpcResponse;
                Object result;

                String url = instance.toUrl();

                try {
                    rpcResponse = httpInvoker.post(request, instance.toUrl());
                    result = castReturnResult(method, rpcResponse);
                } catch (Exception ex) {
                    //故障隔离

                    synchronized (windows) {
                        // 故障的规则统计和隔离，
                        // 每一次异常，记录一次，统计30s的异常数。
                        SlidingTimeWindow window = windows.computeIfAbsent(url, k -> new SlidingTimeWindow());
                        window.record(System.currentTimeMillis());
                        log.debug("instance {} in window with {}", url, window.getSum());
                        if (window.getSum() >= faultLimit) {
                            //30秒内，失败10次，加入到故障隔离中。
                            isolate(instance);
                        }
                    }

                    throw ex;
                }

                synchronized (providers) {
                    //故障恢复
                    if (!providers.contains(instance)) {
                        isolatedProviders.remove(instance);
                        providers.add(instance);
                        log.debug("instance {} is recovered, isolatedProviders={}, providers={}"
                                , instance, isolatedProviders, providers);
                    }
                }

                //过滤器 postfilter
                for (Filter filter : this.context.getFilters()) {
                    Object postResult = filter.postfilter(request, rpcResponse, result);
                    if (postResult != null) {
                        return postResult;
                    }
                }
                return result;
            } catch (Exception ex) {
                if (!((ex.getCause()) instanceof SocketTimeoutException)) {
                    throw ex;
                }
            }
        }
        return null;

    }

    /**
     * 故障隔离
     *
     * @param instance
     */
    private void isolate(InstanceMeta instance) {
        log.debug(" ==> isolate instance: " + instance);

        providers.remove(instance);
        log.debug(" ==> providers = {}", providers);

        isolatedProviders.add(instance);
        log.debug(" ==> isolatedProviders = {}", isolatedProviders);
    }

    /**
     * 返回结果类型转换
     *
     * @param method
     * @param rpcResponse
     * @return
     */
    private static Object castReturnResult(Method method, RpcResponse<?> rpcResponse) {
        if (rpcResponse.isStatus()) {
            Object data = rpcResponse.getData();
            return TypeUtils.castMethodResult(method, data);
        } else {
            RpcException exception = rpcResponse.getEx();
            if (exception != null) {
                log.error("response error.", exception);
                throw exception;
            }
            return null;
        }
    }


}
